package com.dec.kks.etl.query;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

public class HBaseQuery {

    static void putString() throws IOException {
        System.setProperty("hadoop.home.dir","/home/hdfs/bigdata/hadoop-2.7.4");
        Configuration conf = HBaseConfiguration.create();
        Connection conn = ConnectionFactory.createConnection(conf);
        Put put = new Put(Bytes.toBytes("eee"));
        String time = System.currentTimeMillis()+"";
        byte[] tb = Bytes.toBytes(time);

        put.addColumn(Bytes.toBytes("f"),Bytes.toBytes("cl"),tb);
        System.out.println(tb+","+time);
        Table table = conn.getTable(TableName.valueOf("test"));
        table.put(put);

        Get get = new Get(Bytes.toBytes("eee"));
        get.addColumn(Bytes.toBytes("f"),Bytes.toBytes("cl"));
        Result res = table.get(get);
        byte[] v = res.getValue(Bytes.toBytes("f"), Bytes.toBytes("cl"));
        System.out.println(Bytes.toString(v));
    }


    static void putLong() throws Exception {
        System.setProperty("hadoop.home.dir","/home/hdfs/bigdata/hadoop-2.7.4");
        Configuration conf = HBaseConfiguration.create();
        Connection conn = ConnectionFactory.createConnection(conf);
        Put put = new Put(Bytes.toBytes("kkk"));
        long time = System.currentTimeMillis();
        byte[] tb = Bytes.toBytes(time);

        put.addColumn(Bytes.toBytes("f"),Bytes.toBytes("cl"),tb);
        System.out.println(tb+","+time);
        Table table = conn.getTable(TableName.valueOf("test"));
        table.put(put);

        Get get = new Get(Bytes.toBytes("kkk"));
        get.addColumn(Bytes.toBytes("f"),Bytes.toBytes("cl"));
        Result res = table.get(get);
        byte[] v = res.getValue(Bytes.toBytes("f"), Bytes.toBytes("cl"));
        System.out.println(Bytes.toLong(v));

    }

    static void putDouble() throws Exception {
        System.setProperty("hadoop.home.dir","/home/hdfs/bigdata/hadoop-2.7.4");
        Configuration conf = HBaseConfiguration.create();
        Connection conn = ConnectionFactory.createConnection(conf);
        Put put = new Put(Bytes.toBytes("ddd"));
        double d = 23234234.243344d;
        byte[] tb = Bytes.toBytes(d);

        put.addColumn(Bytes.toBytes("f"),Bytes.toBytes("cl"),tb);
        System.out.println(tb+","+d);
        Table table = conn.getTable(TableName.valueOf("test"));
        table.put(put);

        Get get = new Get(Bytes.toBytes("ddd"));
        get.addColumn(Bytes.toBytes("f"),Bytes.toBytes("cl"));
        Result res = table.get(get);
        byte[] v = res.getValue(Bytes.toBytes("f"), Bytes.toBytes("cl"));
        System.out.println(Bytes.toDouble(v));

    }

    static void scan() throws Exception {
        System.setProperty("hadoop.home.dir","/home/hdfs/bigdata/hadoop-2.7.4");
        Configuration conf = HBaseConfiguration.create();
        Connection conn = ConnectionFactory.createConnection(conf);

        Table table = conn.getTable(TableName.valueOf("test"));


        Scan scan = new Scan();
        scan.addColumn(Bytes.toBytes("f"),Bytes.toBytes("cl"));
//        scan.setRowPrefixFilter(Bytes.toBytes("row"));
        ResultScanner rs = table.getScanner(scan);
        try {
            for (Result r = rs.next(); r != null; r = rs.next()) {
                byte[] v = r.getValue(Bytes.toBytes("f"), Bytes.toBytes("cl"));
                System.out.println(Bytes.toString(v));
                System.out.println(Bytes.toLong(v));
                System.out.println(Bytes.toDouble(v));
            }
        } finally {
            rs.close();  // always close the ResultScanner!
        }

    }






    public static void main(String[] args) throws Exception {
        putString();
        putLong();
        putDouble();
        scan();
    }

}
